package storage.multibuffer;

import compiler.absyn.Constant;
import engine.query.ProductScan;
import engine.query.Scan;

import storage.record.TableInfo;
import storage.tx.Transaction;

public class MultiBufferProductScan implements Scan{

	private Scan lhsscan;
	private Scan rhsscan = null;
	private Scan prodscan;
	private TableInfo ti;
	private Transaction tx;
	private int filesize;
	private int chunksize;
	private int nextblknum;
	
	public MultiBufferProductScan(Scan scan, TableInfo ti, Transaction tx)  {
		this.lhsscan = scan;
		this.ti = ti;
		this.tx = tx;
		filesize = tx.size(ti.fileName());
		chunksize = BufferNeeds.bestFactor(filesize);   //should be a initial number, but not zero
		beforeFirst();
	}
	@Override
	public void beforeFirst() {
		// TODO Auto-generated method stub
		nextblknum = 0;
		useNextChunk();
		
	}

	@Override
	public void close() {
		// TODO Auto-generated method stub
		prodscan.close();
	}

	@Override
	public int getInt(String fldname){
		// TODO Auto-generated method stub
		return prodscan.getInt(fldname);
	}

	@Override
	public String getString(String fldname){
		// TODO Auto-generated method stub
		return prodscan.getString(fldname);
	}

	@Override
	public Constant getVal(String fldname){
		// TODO Auto-generated method stub
		return prodscan.getVal(fldname);
	}

	@Override
	public boolean hasField(String fldname) {
		// TODO Auto-generated method stub
		return prodscan.hasField(fldname);
	}

	@Override
	public boolean next()  {
		// TODO Auto-generated method stub
		while(!prodscan.next()) {
			if(!useNextChunk()) {
				return false;
			}
		}
		return true;
	}
	
	private boolean useNextChunk()  {
		if(rhsscan != null) {
			rhsscan.close();
		}
		if(nextblknum >= filesize) {
			return false;
		}
		int end = nextblknum + chunksize -1;
		if(end >= filesize) {
			end = filesize -1;
		}
		rhsscan = new ChunkScan(ti, nextblknum, end, tx);
		lhsscan.beforeFirst();
		prodscan = new ProductScan(lhsscan, rhsscan);
		nextblknum = end + 1;
		return true;
	}

}
